Maybe you can just use CommitOperator with commit.strict-mode.last-safe-snapshot configured in DataEvolutionMergeIntoAction.

Maybe you can just use CommitOperator with commit.strict-mode.last-safe-snapshot configured in DataEvolutionMergeIntoAction.

@JingsongLi Thanks for your advise! I've modified my code.

I mean you can use configure scanned snapshot id to commit.strict-mode.last-safe-snapshot, this snapshot id should be used by MERGE INTO source.

And you don't need to introduce MergeIntoCommitterOperator, just use CommitterOperator, commit.strict-mode.last-safe-snapshot will solve your FilterAndCommit problem.

I mean you can use configure scanned snapshot id to commit.strict-mode.last-safe-snapshot, this snapshot id should be used by MERGE INTO source.

And you don't need to introduce MergeIntoCommitterOperator, just use CommitterOperator, commit.strict-mode.last-safe-snapshot will solve your FilterAndCommit problem.

@JingsongLi Thanks for your explaination! I've set the commit.strict-mode.last-safe-snapshot as the latestSnapshotId in the action's constructor.

I introduce the new committer operator wrapping a CommitterOperator to check the received committables, so that we could throw and error or drop partition index if merge into action modified some global-indexed columns.

An alternative approach is to introduce a new operator before committer e.g. UpdateCheckerOperator. Which one do you prefer, I'm willing to modify my code.

@steFaiz Sounds good to me.

@steFaiz Sounds good to me.

@JingsongLi Appreciate for it! Do you mean I should introduce a UpdateCheckerOperator or keep the new committer(just resolve the conflict)?

@steFaiz Sounds good to me.

@JingsongLi Appreciate for it! Do you mean I should introduce a UpdateCheckerOperator or keep the new committer(just resolve the conflict)?

introduce a UpdateCheckerOperator

+1